/**
 * Copyright 2008-2009. Chongqing Communications Industry Services Co.,Ltd Information Technology Branch. All rights
 * reserved. <a>http://www.cqcis.com</a>
 */
package com.cqcis.uip.base.framework.core.impl;

import java.util.Set;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.common.IoSession;

import com.cqcis.uip.base.framework.UIPServiceException;
import com.cqcis.uip.base.framework.base.AbstractConnectorSupport;
import com.cqcis.uip.base.framework.base.adaptor.AbstractSocketService;
import com.cqcis.uip.base.framework.common.Task;
import com.cqcis.uip.base.framework.core.RegisteredService;
import com.cqcis.uip.base.framework.core.RegisteredServiceContext;
import com.cqcis.uip.base.framework.core.TaskContext;
import com.cqcis.uip.base.framework.core.TaskContextSessionStrategy;
import com.cqcis.uip.base.framework.utils.ConcurrentTaskQueue;
import com.cqcis.uip.base.framework.utils.SessionUtil;
import com.cqcis.uip.base.framework.utils.TaskUtil;
import com.cqcis.uip.base.framework.utils.ConcurrentTaskQueue.OfferedListener;

/**
 * task上下文定义实现
 * 
 * @author huangmy
 * @date   2008-8-26
 */
public abstract class AbstractTaskContext<T> implements TaskContext, OfferedListener {
	private static Log log = LogFactory.getLog(AbstractTaskContext.class);
	private final static String CONNECTBEFOREWRITE_SESSION = "CONNECTBEFOREWRITE_SESSION_" + AbstractTaskContext.class.getName();
	public final static String DONT_REMOVETASK_SESSION = "DONT_REMOVETASK_SESSION_" + AbstractTaskContext.class.getName();
	public final static String TASK_CURRENT_BIND_SESSION = "TASK_CURRENT_BIND_SESSION_" + AbstractTaskContext.class.getName();
    private RegisteredServiceContext serviceContext; //  服务上下文
    private Task task; // 关联的task
    private WriteRecord writeRecord; // 写操作记录
    private ConcurrentTaskQueue<T> messageQueue = new ConcurrentTaskQueue<T>();
    private Thread contextRelaThread;// 关联的处理线程
    private boolean keeplive; // 是否长连接
    private boolean connectBeforeWrite; // 写前新建连接
    private TaskContextSessionStrategy<Object> taskContextSessionStrategy; // 新建session
    private Set<Task> requestSessionTasks; // 请求session的task
    private ConcurrentTaskQueue<Object> requestSessions;
        
    /**
     * 默认构建
     * 
     * @param serviceContext
     */
    public AbstractTaskContext(RegisteredServiceContext serviceContext, final Task task) {
    	this.serviceContext = serviceContext;
    	this.task = task;
    	
    	RegisteredService rst = serviceContext.getRegisteredService();
    	if (rst instanceof AbstractConnectorSupport) {
    		AbstractConnectorSupport rs = (AbstractConnectorSupport)rst;
        	this.keeplive = rs.isKeepAlive();
    	}
    	
    	this.messageQueue.addOfferedListener(this);
    }
    
    public void offered(Object t) {
		if (this.connectBeforeWrite) {    				
			IoSession newSession = (IoSession)this.task.getAttribute(TASK_CURRENT_BIND_SESSION);
			newSession.setAttribute(DONT_REMOVETASK_SESSION, Boolean.TRUE);
		}
	}
    
	public RegisteredServiceContext getRegisteredServiceContext() {		
		return this.serviceContext;
	}
	
	protected void checkOperationValid() {
		RegisteredService serviceType = this.serviceContext.getRegisteredService();
		if (!(serviceType instanceof AbstractSocketService)) {
			throw new UnsupportedOperationException("非SOCKET CLIENT类型服务不支持该方法");
		}
	}
	
	protected Task getTask() {
		return this.task;
	}
	
	public void setConnectBeforeWrite(boolean connectNewSession) {
		if (this.keeplive) { //  必须是短连接才有效
			return;
		}
		
		if (this.keeplive && connectNewSession) {
			throw new UIPServiceException("Connect New Session Before Write, the client must be not keeplive, " +
					"设置发送数据前新建链接，要求链接类型设置为短链接!");
		}
		
		getSessionForTask().setAttribute(DONT_REMOVETASK_SESSION, Boolean.TRUE);
		this.connectBeforeWrite = connectNewSession;
		
		// clear all the session
		try {
			//((SocketSessionImpl)getSessionForTask()).getChannel().close();
			getSessionForTask().close().join();
		} catch (Exception e) {
		}// close
	}
	
	public void write(Object o) throws Exception {
		checkOperationValid();// check
		IoSession session = getSessionForTask();
		if (this.connectBeforeWrite && session != null) { // 写前建立连接
			IoSession newSession = createAloneNewSession(session);			
			SessionUtil.setCurrentTask(newSession, this.task);
			this.task.setAttribute(TASK_CURRENT_BIND_SESSION, newSession);
			newSession.setAttribute(CONNECTBEFOREWRITE_SESSION, Boolean.TRUE);
			TaskUtil.setCurrentSession(this.task, newSession);
		} else if (session != null && session.isClosing()) {
			throw new UIPServiceException("Socket Session is Closed");
		}
		
		if (this.writeRecord != null) {
			this.writeRecord.write(o);
		}
		o = doBeforeWrite(o); // 写前处理
		doWrite(o); // 同步阻塞写
	}
	
	/**
	 * 获取session
	 * 
	 * @return
	 */
	protected IoSession getSessionForTask() {
		return this.task.getSession();
	}
		
	/**
	 * 创建独立的新链接
	 * 
	 * @param parentSession
	 * @return
	 */
	private IoSession createAloneNewSession(IoSession parentSession) {
		IoSession session = null;
		try {
		    this.requestSessionTasks.add(this.task);
		    this.taskContextSessionStrategy.newSession();
		    long timeout = TaskUtil.getTimeout(this.task) + TaskUtil.getTimeout(this.task)/5;
		    // wait for session
		    if (log.isDebugEnabled()) {
				log.debug("Ask Create Alone Session: waitTimeout = " + timeout);
			}
		    
		    Object t = this.requestSessions.poll(timeout);
		    if (t == null) {
		    	throw new UIPServiceException("Failed to get session.");
		    } else if (t instanceof Throwable) {
		    	throw new UIPServiceException((Throwable)t);
		    }
		    
		    session = (IoSession)t;
		} finally {
			this.requestSessionTasks.remove(this.task);
		}
		
		if (log.isDebugEnabled()) {
			log.debug("Create Alone Session: " + session + ", taskId: " + this.task.getTaskId());
		}
		
		return session;
	}
	
	/**
	 * 写
	 * 
	 * @param o
	 */
	protected void doWrite(Object o) throws Exception {
		TaskUtil.writeSync(task, o); // 同步阻塞写
	}

	public Object read() throws Exception {
		checkOperationValid();// check		
		IoSession session = getSessionForTask();
		if (this.connectBeforeWrite && session != null) {
			Object obj = doRead();// 读取
			try {
				IoSession newSession = TaskUtil.getCurrentSession(this.task);
				Object cbws = newSession.getAttribute(CONNECTBEFOREWRITE_SESSION);
				if (cbws != null) {		
					newSession.setAttribute(DONT_REMOVETASK_SESSION, Boolean.TRUE);
					//((SocketSessionImpl)newSession).getChannel().close(); // 关闭
					newSession.close().join();
				}
			} catch (Exception e) {
			} finally {
				if (obj == null) {
					throw new UIPServiceException("Read Object is null.");
				}
			}
			
			return obj;
		} else if (session != null && session.isClosing()) {
			throw new UIPServiceException("Socket Session is Closed");
		}
		
		Object obj = doRead();// 读取
		if (obj == null) {
			throw new UIPServiceException("Read Object is null.");
		}
		return obj;
	}
	
	public static interface WriteRecord {
		/**
		 * 记录写操作
		 * 
		 * @param o
		 */
		public void write(Object o);
	}

	public void setWriteRecord(WriteRecord writeRecord) {
		this.writeRecord = writeRecord;
	}
	
	/**
	 * 读取
	 */	
	protected T doRead() {
		long timeout = TaskUtil.getTimeout(this.task) + TaskUtil.getTimeout(this.task)/5;
		
		if (log.isDebugEnabled()) {
			log.debug("doRead: waitTimeout = " + timeout);
		}
		
		return this.messageQueue.poll(timeout);
	}
	
	/**
	 * 添加信息
	 * 
	 * @param message
	 */
	public void offerMessage(T message) {
		this.messageQueue.offer(message);
	}
	
	/**
	 * 写数据前处理, 一般情况下原样返回
	 * 
	 * @param o
	 * @return
	 */
	public abstract Object doBeforeWrite(Object o);

	/**
	 * 中断执行
	 * 
	 */
	public void interrupt() {
		Thread t = getContextRelaThread();
		if (t != null) {
			t.interrupt();
		}
	}
	
	/**
	 * 是否中断，如果尚未执行，也认为是中断
	 * 
	 * @return
	 */
	public boolean isInterrupted() {
		Thread t = getContextRelaThread();
		if (t != null) {
			return t.isInterrupted();
		}
		return true;
	}
	
	/**
	 * 关联线程
	 * 
	 * @return
	 */
	public Thread getContextRelaThread() {
		return contextRelaThread;
	}

	/**
	 * 设置关联线程
	 * 
	 * @param contextRelaThread
	 */
	public void setContextRelaThread(Thread contextRelaThread) {
		this.contextRelaThread = contextRelaThread;
	}

	public void setRequestSessionTasks(Set<Task> requestSessionTasks) {
		this.requestSessionTasks = requestSessionTasks;
	}

	public void setRequestSessions(ConcurrentTaskQueue<Object> requestSessions) {
		this.requestSessions = requestSessions;
	}

	public void setTaskContextSessionStrategy(
			TaskContextSessionStrategy<Object> taskContextSessionStrategy) {
		this.taskContextSessionStrategy = taskContextSessionStrategy;
	}
}
